package com.codejiwei.flink.sql;

import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * author: codejiwei
 * date: 2023/6/8
 * desc:
 **/
public class Flink_Table_SQL_API3 {
    public static void main(String[] args) {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

//        tableEnv.useCatalog("custom_catalog");

        //source
        tableEnv.executeSql("create temporary table sourceTableC(col1 String, col2 bigint, col3 timestamp) " +
                "with ('connector' = 'datagen', " +
                "'rows-per-second' = '10', " +
                "'fields.col1.length' = '5', " +
                "'fields.col2.min' = '1001', " +
                "'fields.col2.max' = '1099'" +
                ")");


        //sink
        tableEnv.executeSql("create table sinkTable (sinkCol1 String, sinkCol2 bigint, sinkCol3 timestamp(3)) with ('connector' = 'print')");

        //insert
//        tableEnv.executeSql("insert into sinkTable select col1, col2, col3 from sourceTableC");
        tableEnv.sqlQuery("select col1, col2, col3 from sourceTableC").executeInsert("sinkTable");


    }
}
